package com.stone.zk.lock.app.service;

import com.stone.zk.lock.config.ZookeeperConfig;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.Collections;
import java.util.List;

/**
 * 基于Zookeeper实现分布式锁
 *
 * @author Mr_wenpan@163.com 2021/8/2 2:01 下午
 */
@Service
public class ZookeeperLockService implements AutoCloseable, Watcher {


    private final ZooKeeper zk;

    /**
     * todo 创建的临时节点，这里并发下会有线程安全问题，待优化
     */
    private String zkNode;

    public ZookeeperLockService(ZookeeperConfig config) throws IOException {
        zk = new ZooKeeper(config.getConnectString(), config.getSessionTimeout(), this);
    }

    public boolean getLock(String key) {

        try {
            //首先，判断业务节点（持久节点）是否存在，不存在就创建
            Stat stat = zk.exists("/" + key, false);
            if (stat == null) {
                zk.create("/" + key, key.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
            // 创建临时有序节点
            zkNode = zk.create("/" + key + "/" + key + "_", key.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            // 排序，选取最小节点，即查看业务节点下所有节点，排序，选第一个
            List<String> childrenNodes = zk.getChildren("/" + key, false);
            Collections.sort(childrenNodes);
            // 获取序号最小的（第一个）子节点
            String firstNode = childrenNodes.get(0);
            // 判断当前节点是否是第一个，即是否应该获取到锁
            if (zkNode.endsWith(firstNode)) {
                return true;
            }

            // 如果当前节点不能获取到锁，则需要添加监听器。
            // lastNode，遍历过程中，保存当前节点的前一个节点，并添加监听器
            String lastNode = firstNode;
            for (String node : childrenNodes) {
                if (zkNode.endsWith(node)) {
                    zk.exists("/" + key + "/" + lastNode, true);
                    break;
                } else {
                    lastNode = node;
                }
            }

            // 锁定当前对象
            synchronized (this) {
                wait();
            }

            return true;
        } catch (Exception e) {
            e.printStackTrace();
        }

        return true;
    }

    @Override
    public void close() throws Exception {
        zk.delete(zkNode, -1);
        zk.close();
    }

    /**
     * 监听器方法，监听到删除节点操作(即释放锁)，唤醒下一个节点（线程）
     *
     * @param event 监听事件
     */
    @Override
    public void process(WatchedEvent event) {
        if (event.getType() == Event.EventType.NodeDeleted) {
            synchronized (this) {
                notify();
            }
        }
    }

}

